package alfred.camel;

import alfred.mdp.client.ClientTimeoutException;
import alfred.mdp.tsp.client.TitanicClient;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangeTimedOutException;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.impl.DefaultAsyncProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZContext;
import org.zeromq.ZDispatcher;
import org.zeromq.ZMsg;

import java.util.concurrent.TimeUnit;

/**
 */
public class TitanicProducer extends DefaultAsyncProducer {
    private static final Logger LOG = LoggerFactory.getLogger(TitanicProducer.class);

    private final String serviceName;
    private final TimeUnit timeUnit;
    private final long timeout;
    private final TitanicClient client;

    public TitanicProducer(AlfredEndpoint endpoint, ZContext zmqContext, ZDispatcher dispatcher, String serviceName, TimeUnit timeUnit, long timeout) {
        super(endpoint);
        this.serviceName = serviceName;
        this.timeUnit = timeUnit;
        this.timeout = timeout;
        this.client = new TitanicClient(zmqContext, dispatcher, endpoint.getComponent().getClientsUri());
    }

    @Override
    public boolean process(Exchange exchange, AsyncCallback callback) {
        LOG.debug("Sending message with Titanic client.");
        try {
            ZMsg body = exchange.getIn().getBody(ZMsg.class);
            if (body == null) {
                LOG.warn("Exchange body must be not null, Exchange:{}", exchange);

            } else {
                long startTime = System.currentTimeMillis();

                String uuid = client.request(serviceName, body, timeUnit, timeout);

                ZMsg response;
                while ((response = client.reply(uuid, timeUnit, getRemainingTime(startTime, timeUnit, timeout))) == null);

                client.close(uuid, timeUnit, getRemainingTime(startTime, timeUnit, timeout));

                exchange.getOut().setBody(response);
            }

        } catch (ClientTimeoutException e) {
            exchange.setException(new ExchangeTimedOutException(exchange, timeout));

        } catch (Exception ex) {
            LOG.error("Error occurred while sending message with Titanic client.", ex);
            throw new RuntimeCamelException(ex);

        } finally {
            LOG.debug("Message sent with Titanic client.");

        }
        callback.done(true);
        return true;
    }

    private long getRemainingTime(long startTime, TimeUnit timeUnit, long timeout) {
        return timeUnit.convert(TimeUnit.MILLISECONDS.convert(timeout, timeUnit) - (System.currentTimeMillis() - startTime), TimeUnit.MILLISECONDS);
    }

}
